package sync.function;

import org.apache.flink.streaming.api.functions.ProcessFunction;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: net.bwie.realtime.sync.function
 * @Author: zhangleqing
 * @CreateTime: 2025-08-30  09:46
 * @Description: TODO
 * @Version: 1.0
 */
public class LogSplitFunction extends ProcessFunction<String, String> {
    /**
     * 侧边流输出时标记outputtag实例
     *
     * @param s
     * @param context
     * @param collector
     * @throws Exception
     */
    private final OutputTag<String> orderBaseTag;
    private final OutputTag<String> refundDetailTag;
    private final OutputTag<String> logisticsInfoTag;
    private final OutputTag<String> consumerBehaviorTag;

    public LogSplitFunction(OutputTag<String> orderBaseTag,
                            OutputTag<String> refundDetailTag,
                            OutputTag<String> logisticsInfoTag,
                            OutputTag<String> consumerBehaviorTag) {

        this.orderBaseTag = orderBaseTag;
        this.refundDetailTag = refundDetailTag;
        this.logisticsInfoTag = logisticsInfoTag;
        this.consumerBehaviorTag = consumerBehaviorTag;
    }

    /*
    jsonObject.getLong("ts_ms") 中的 ts_ms，是 Debezium 在捕获 MySQL 变更（Insert/Update/Delete）时，
    自动附加的元数据字段，而非你 MySQL 表中的业务字段。
     */

    @Override
    public void processElement(String value,
                               Context ctx,
                               Collector<String> out) throws Exception {
        // todo 解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        // 如果json不是空值
        if (null != jsonObject){
            // todo 过滤出 order_base 表的数据
            if (jsonObject.getJSONObject("source").getString("table").equals("order_base")){
                JSONObject after = jsonObject.getJSONObject("after");

                Long ts_ms = jsonObject.getLong("ts_ms");
                after.put("ts_ms", ts_ms);
                // 侧边流输出
                ctx.output(orderBaseTag, after.toJSONString());
                System.out.println(after.toJSONString());
            }

            // todo 过滤出 refund_detail 表的数据
            if (jsonObject.getJSONObject("source").getString("table").equals("refund_detail")){
                JSONObject after = jsonObject.getJSONObject("after");

                Long ts_ms = jsonObject.getLong("ts_ms");
                after.put("ts_ms", ts_ms);
                // 侧边流输出
                ctx.output(refundDetailTag, after.toJSONString());
                System.out.println(after.toJSONString());
            }

            // todo 过滤出 logistics_info 表的数据
            if (jsonObject.getJSONObject("source").getString("table").equals("logistics_info")){
                JSONObject after = jsonObject.getJSONObject("after");

                Long ts_ms = jsonObject.getLong("ts_ms");
                after.put("ts_ms", ts_ms);
                // 侧边流输出
                ctx.output(logisticsInfoTag, after.toJSONString());
                System.out.println(after.toJSONString());
            }

            // todo 过滤出 consumer_behavior 表的数据
            if (jsonObject.getJSONObject("source").getString("table").equals("consumer_behavior")){
                JSONObject after = jsonObject.getJSONObject("after");

                Long ts_ms = jsonObject.getLong("ts_ms");
                after.put("ts_ms", ts_ms);
                // 侧边流输出
                ctx.output(consumerBehaviorTag, after.toJSONString());
                System.out.println(after.toJSONString());
            }
        }

    }
}
